package com.freez.spark.core.operation;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: MapOperation.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:04
 * Description: 对map,mapToPair,flatMap,flatMapToPair进行演示
 */
public class MapOperation {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("SparkCore-MapOperation").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        //数据
        List<Tuple2<String, Integer>> dataList = new ArrayList<>();
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 11));
        dataList.add(new Tuple2<String, Integer>("2020-05-02", 4));
        dataList.add(new Tuple2<String, Integer>("2020-05-02", 7));
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 4));
        dataList.add(new Tuple2<String, Integer>("2020-05-01", 10));
        JavaPairRDD<String, Integer> aRdd = jsc.parallelizePairs(dataList);
        JavaRDD<String> bRdd = jsc.parallelize(Arrays.asList("hello world", "hello hi"));
        /**
         * map 返回String
         */
        JavaRDD<String> map = bRdd.map(new Function<String, String>() {
            @Override
            public String call(String s) throws Exception {
                return s;
            }
        });
        map.foreach(new VoidFunction<String>() {
            @Override
            public void call(String s) throws Exception {
                System.out.println(s);
            }
        });
        /**
         * mapToPair
         */
        JavaPairRDD<Integer, String> bKVRDD = bRdd.mapToPair(new PairFunction<String, Integer, String>() {
            @Override
            public Tuple2<Integer, String> call(String s) throws Exception {
                return new Tuple2<>(1, s);
            }
        });
        bKVRDD.foreach(new VoidFunction<Tuple2<Integer, String>>() {
            @Override
            public void call(Tuple2<Integer, String> s) throws Exception {
                System.out.println(s);
            }
        });
        /**
         * flatMap 返回Iterator
         */
        JavaRDD<String> flatMapRDD = bRdd.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String s) throws Exception {
                return Arrays.asList(s.split(" ")).iterator();
            }
        });
        flatMapRDD.foreach((VoidFunction<String>) s -> System.out.println(s));
        /**
         * flatMapToPair
         */
        bRdd.flatMapToPair(new PairFlatMapFunction<String, String, String>() {
            @Override
            public Iterator<Tuple2<String, String>> call(String s) throws Exception {
                String[] split = s.split(",");
                String[] split1 = split[0].split(" ");
                List<Tuple2<String, String>> list = new ArrayList<>();
                list.add(new Tuple2<>(split1[0], split1[1]));
                return list.iterator();
            }
        });
        jsc.close();

    }
}